package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.Constants;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.apollo.ApolloConfig;
import com.aotain.apollo.ServerInfo;
import com.aotain.project.apollo.SiteEvaluateMain;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.storm.AbstractBolt;


/**
 * @ClassName: BaseStatSumBolt 
 * @Description:  全流量统计 针对目的IP，进行各时间维度的数据统计，结果存HBASE
 * 					  并且将计算后的结果下传，用于IDC机房的数据统计
 * @author Turk
 * @date 2015年 
 */
public class BaseStatSumBolt extends AbstractBolt {

	/**
	 * 配置指定检测IP
	 */
//	HashMap<String,String> mIPs = new HashMap<String,String>();
	
	//HashMap<String,Long> mapSourceArea = new HashMap<String,Long>();
	
	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	/**
	 * 流量来源统计
	 */
	HashMap<String,HashMap<String,Long>> mapSourceArea 
		= new HashMap<String,HashMap<String,Long>>();
	
	//private String CurrentDate = "";
	
	HashMap<String,ServerInfo> mapServers = null;
	
	
	//HashMap<String,Integer> ScanIP =  new HashMap<String,Integer>();
	
	HashMap<String,Long> DestIPPV = new HashMap<String,Long>();
	
	
	
	long TotalPV = 0L;
	
	long _updateTime = 0L;
	
//	HashSet<String> _ImportIPSet = new HashSet<String>();
	
	
	public BaseStatSumBolt(){
	
	}
	
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare(new Fields("IDC","TYPE","PV","PACKET"));
		declarer.declareStream("DELAY", new Fields("IDC","IP","TYPE","PV","PACKET"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		//
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,300);
		return conf;
	}


	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		
	}
	
	
	private void countInLocal(Tuple tuple) {
		
		
		String DestIP = tuple.getString(0);
//		int destport = Integer.parseInt(tuple.getString(1));
		
		String idc = tuple.getStringByField("idcno");
		
//		String sourceip = tuple.getString(2);
//		String sport = tuple.getString(3);
//		System.err.println("gis64========" + tuple.getString(4));
		byte[] gisbytes =  Base64.decodeBase64(tuple.getString(4));
		String gisstr = new String(gisbytes);
		//destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#" + Sourcelon + "," + Sourcelat + "#" + sourceAreaCountry;
		//destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#" + Sourcelon + "," + Sourcelat + "#" + sourceAreaCountry + "#" + sourceAreaId + "#" +areaName;
		String[] gisArray = gisstr.split("#",-1);
//		String destAreaName = gisArray[0]; //如果地市为空，精确到国家
//		String DestGeo = gisArray[1];
//		String sourceAreaName = gisArray[2];  //如果地市为空，精确到国家
//		String SourceGeo = gisArray[3];
//		String sourceAreaCountry = gisArray[4];
		String sourceProvinceID =  gisArray[5];
		String sourceProvinceName =  gisArray[6]; //如果省为空，精确到国家
		

		//String dip = tuple.getString(0);
		
		String key = String.format("%s|%s", idc, DestIP);

		if(DestIPPV.containsKey(key))
		{
			long dipPV = 0;
			dipPV = DestIPPV.get(key) + 1;
			DestIPPV.put(key, dipPV);
		}
		else
		{
			DestIPPV.put(key, 1L);
		}
		
		TotalPV++;
		
		
		//区域分布统计
		String sourceKey = sourceProvinceID + "@" + sourceProvinceName;
		if(mapSourceArea.containsKey(key))
		{
			HashMap<String,Long> sourceArea = mapSourceArea.get(key);
			
			if(sourceArea.containsKey(sourceKey))
			{
				long sourcePV = 0; 
				sourcePV = sourceArea.get(sourceKey) + 1;
				sourceArea.put(sourceKey, sourcePV);
			}
			else
			{
				sourceArea.put(sourceKey, 1L);
			}	
		}
		else
		{
			HashMap<String,Long> sourceArea = new HashMap<String,Long>();
			sourceArea.put(sourceKey, 1L);
			mapSourceArea.put(key, sourceArea);
		}
		
	}
	
	private void emitCountingData(Tuple input) {
		
		
		Date date = new Date();
		SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmm");
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
		//SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH");
		SimpleDateFormat sdf_min  = new SimpleDateFormat("mm");
		String dateStr_min = sdf_min.format(date);
		
		String dateStr_h = sdf_h.format(date);
		String dateStr_d = sdf_d.format(date);		
		
		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
		
		SiteEvaluateMain evaluate = new SiteEvaluateMain(zooserver);
		boolean IsTime = false;
		if(System.currentTimeMillis() - _updateTime >= 1800*1000L)
		{
			IsTime = true;
			_updateTime = System.currentTimeMillis();
			//1小时获取一次新的入库IP配置
//			Slave s = new Slave(zooserver);
//			_ImportIPSet = s.GetIPFilterForLog();
		}
		
		for(String key : DestIPPV.keySet()) {
			
//			if(mIPs.size() > 0 && !mIPs.containsKey(dkey))
//			{
//				//_collector.ack(input);
//				continue;
//			}
			
			String IDC = key.split("\\|",-1)[0];
			String destIP = key.split("\\|",-1)[1];
			
			long avgPvalue = DestIPPV.get(key);
			String cf = "cf";
			
			//######  5Min  ##### 
//			if(_ImportIPSet!=null && _ImportIPSet.size() > 0 && _ImportIPSet.contains(destIP))
			{//满足IP配置条件 入库 减缓入库压力
				int nMin5 = (Integer.parseInt(dateStr_min) / 5) * 5;
				String sMin5 = String.format("%02d", nMin5);   
				String dateStr5 = dateStr_h + sMin5;
				String rowkey5 = String.format("%s_%s", destIP, dateStr5);
				
				
				String tbName5 = ApolloProperties.SDS_SESSION_STAT_5MIN;
				hbaseInstance.Add(tbName5, rowkey5, cf,  "DESTIP", destIP);
				hbaseInstance.Add(tbName5, rowkey5, cf,  "REPORTTIME", dateStr5);
				hbaseInstance.Add(tbName5, rowkey5, cf,  "PV", (long)avgPvalue);

			}
			
			
			SimpleDateFormat dfDay = new SimpleDateFormat("yyyyMMdd");
			Date dStartTimeDay = new Date();
			String strDateDay = dfDay.format(dStartTimeDay);
			
			Calendar c = Calendar.getInstance();
			c.add(Calendar.DATE, 1);// 1 minute ago
			String strEndDay = dfDay.format(c.getTime());
			
			if(IsTime)
			{
				//IP
//				if(_ImportIPSet!=null && _ImportIPSet.size() > 0 && _ImportIPSet.contains(destIP))
				{
					evaluate.EvaluateFunction(destIP, strDateDay, strEndDay, avgPvalue);
					
					String statRowKeyD = String.format("%s_%s", destIP, strDateDay);
			    	String abnStatTbNameDay = ApolloProperties.SDS_ABN_SESS_STAT_D;

			    	if(mapServers != null && mapServers.containsKey(destIP))
			    	{
			    		ServerInfo serinfo = mapServers.get(destIP);
			    		hbaseInstance.Add(abnStatTbNameDay, statRowKeyD, cf, "ACCESSTYPE", serinfo.getAccessType());
			    		hbaseInstance.Add(abnStatTbNameDay, statRowKeyD, cf, "LOCATION", serinfo.getServerAddress());
			    	}
			    	else
			    	{
			    		hbaseInstance.Add(abnStatTbNameDay, statRowKeyD, cf, "ACCESSTYPE", "NONE");
			    		hbaseInstance.Add(abnStatTbNameDay, statRowKeyD, cf, "LOCATION", "NONE");
			    	}
				}
				
				
			}
			
	    	collector.emit(new Values(IDC,"PV",avgPvalue,0L));
	    	collector.emit("DELAY",new Values(IDC,destIP, "PV", avgPvalue,0L));
		}
		
		DestIPPV.clear();
		IsTime = false;
		
		
		
		for(Entry<String, HashMap<String,Long>> destIPMap : mapSourceArea.entrySet())
		{
			String destIP = destIPMap.getKey();
			HashMap<String,Long> sourceAreas = destIPMap.getValue();
			
			String tbName = ApolloProperties.SDS_SESSION_AREA_D;
			for(Entry<String, Long> sourceArea : sourceAreas.entrySet())
			{
				String sourceKey = sourceArea.getKey();
				String sourceID = sourceKey.split("@",-1)[0];
				String sourceName = sourceKey.split("@",-1)[1];
				
				String rowkey = String.format("%s_%s_%s",destIP, dateStr_d, sourceID);
				String cf = "cf";
				
				hbaseInstance.Add(tbName, rowkey, cf,  "REPORTTIME", dateStr_d);
				hbaseInstance.Add(tbName, rowkey, cf,  "DESTIP", destIP);
				hbaseInstance.Add(tbName, rowkey, cf,  "SOURCEAREA", sourceName);
				hbaseInstance.incerment(tbName, rowkey, "cf:PV", sourceArea.getValue());
			}
			
		}
		
		mapSourceArea.clear();
		
		evaluate.CloseConnection();
		
	}


	@Override
	public void execute(Tuple input) {
		// TODO Auto-generated method stub
		try {
			
			if(isTickTuple(input)) {//满足时间条件
				Logger.getLogger(BaseStatSumBolt.class).info(String.format("###EMIT BaseStatSumBolt START [%s]",input.getSourceStreamId()));
				emitCountingData(input);  //计算，入库
				Logger.getLogger(BaseStatSumBolt.class).info(String.format("###EMIT BaseStatSumBolt END [%s]",input.getSourceStreamId()));
				collector.ack(input);
			} else {//不满足时间条件
				countInLocal(input); //数据记录 
				collector.ack(input);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			Logger.getRootLogger().error(e);
			throw new FailedException("BaseStatSumBolt Exception",e);
		}	
	}

	
	public boolean isTickTuple(Tuple tuple) {
		return tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID) && tuple.getSourceStreamId().equals(
		        Constants.SYSTEM_TICK_STREAM_ID);   
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		String dbJson = stormConf.get("dbjson_").toString(); //oracle
		ApolloConfig ap = new ApolloConfig(dbJson);
		//获取到服务器配置信息	
		mapServers = ap.ServerInfos();
//		_updateTime = System.currentTimeMillis();
//		Slave s = new Slave(zooserver);
//		_ImportIPSet = s.GetIPFilterForLog();
//		String msg = String.format("INIT BassStatSumBolt");
//		s.SendAlarmLog("INFO", msg);
	}
}
